/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.common.testhelper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.testhelper.FaultManager.FaultHelper.InjectionInfo;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventHandler;
import com.nhncorp.pleiades.client.LockEventHandlerFactory;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClientFactory;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.client.LockService.LockMode;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class FaultManager {
  private static final Log LOG = LogFactory.getLog(FaultManager.class);
  private final static String LOCK_PATH = "/fault_injection";

  HashMap<Class<?>, HashMap<String, Fault>> classFaultMap 
      = new HashMap<Class<?>, HashMap<String, Fault>>();
  LockService lockService;

  FaultManager(NConfiguration conf) throws IOException {
    initPleiadesLockClient(conf);
  }

  public static FaultManager create(NConfiguration conf) throws IOException {
    FaultManager fm = new FaultManager(conf);
    fm.load();
    return fm;
  }

  public void close() {
    lockService.close();
    lockService = null;
  }

  public Fault getFault(Class<?> c, String methodName) throws IOException {
    if (lockService == null) {
      throw new IOException("FaultManager is not initialized");
    }

    HashMap<String, Fault> faultMap = null;
    synchronized (classFaultMap) {
      faultMap = findFaultMap(c);
    }

    if (faultMap == null) {
      return null;
    }

    return faultMap.get(methodName);
  }

  public void clearAllFaults() throws IOException, InterruptedException {
    lockService.deleteNode(LOCK_PATH, true);
    
    Condition cond = new Condition() {
      boolean isSatisfied() {
        return classFaultMap.isEmpty();
      }
    };

    waitUntil(cond, 10000);
  }

  public void clear(final Class<?> c, final String methodName) throws IOException, InterruptedException {
    if (lockService == null) {
      throw new IOException("FaultManager is not initialized");
    }

    String faultLockPath = FaultHelper.createLockPath(c, methodName);
    lockService.setContents(LOCK_PATH + "/" + faultLockPath, "");

    Condition cond = new Condition() {
      boolean isSatisfied() throws IOException {
        return getFault(c, methodName) == null;
      }
    };
    
    waitUntil(cond, 5000);
  }

  /**
   * Pleiades에 fault 정보를 주입한다. 
   * 이 함수의 수행은 fault를 주입한 FaultManager에 해당 fault가 이벤트로써 수신될 때까지
   * blocking한다. 하지만, 이 blocking이 풀린다고 해서 이 fault를 수신해야 하는 
   * 모든 FaultManager가 방금 주입한 fault를 성공적으로 수신한 것은 아니다. 
   * 따라서, 모든 FaultManager가 fault를 수신할 때까지 약간의 sleep을 한다. 
   * @param c
   * @param methodName
   * @param faultClass
   * @throws IOException
   * @throws InterruptedException 
   */
  public void inject(final Class<?> c, final String methodName, final Class<? extends Fault> faultClass) throws IOException, InterruptedException {
    if (lockService == null) {
      throw new IOException("FaultManager is not initialized");
    }

    String faultLockPath = FaultHelper.createLockPath(c, methodName);

    lockService.setContents(LOCK_PATH + "/" + faultLockPath
        , faultClass.getName().replaceAll("\\.", "_"));


    Condition cond = new Condition() {
      boolean isSatisfied() throws IOException {
        Fault fault = null;
        return (fault = getFault(c, methodName)) != null && fault.getClass().equals(faultClass);
      }
    };
    
    waitUntil(cond, 5000);
  }

  private void load() throws IOException {
    Map<String, LockMetaData> dirMap = lockService.readDir(LOCK_PATH, true);

    if (dirMap == null) {
      return;
    }

    for (Entry<String, LockMetaData> entry : dirMap.entrySet()) {
      String fullPath = entry.getValue().getLockId();
      String classAndMethodName = extractClassAndMethodName(fullPath);
      String[] pathElements = classAndMethodName.split("/");

      if (pathElements.length != 2) {
        continue;
      }

      InjectionInfo info = FaultHelper.convertAsInfo(fullPath, lockService.getContents(fullPath));

      if (info.fault != null) {
        LOG.debug("load fault : " + info);
        internalInject(info.c, info.methodName, info.fault);
      }
    }
  }

  private String extractClassAndMethodName(String fullPath) {
    return fullPath.substring(fullPath.indexOf(LOCK_PATH) + LOCK_PATH.length() + 1);
  }

  private void initPleiadesLockClient(NConfiguration conf) throws IOException {
    PleiadesUser user = new PleiadesUser(conf.get("pleiades.service.name", "neptune_test"),
        "fault_manager");
    user.setSessionDisable(false);
    PleiadesConfiguration pleiadesConf = new PleiadesConfiguration();
    lockService = LockServiceClientFactory.newInstance(user, pleiadesConf);
    if (lockService == null) {
      throw new IOException("LockService is not initialized");
    }
    
    LockEventHandler handler = lockService.getEventHandler();
    
    if (handler == null) {
      handler = LockEventHandlerFactory.newInstance(pleiadesConf);
      handler.addLockEventMonitor(createEventMonitor(pleiadesConf));
      lockService.setEventHandler(handler);
    } else {
      handler.addLockEventMonitor(createEventMonitor(pleiadesConf));
    }
  }

  private LockEventMonitor createEventMonitor(PleiadesConfiguration conf) {
    return new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, LOCK_PATH);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.NODE_REMOVED);
        lockEvents.add(LockService.Events.CONTENTS_CHANGED);
        filter.setLockEventList(lockEvents);

        return filter;
      }
      
      public void notifyEvent(LockService.Events event, LockEventData eventData) {
        try {
          String path = eventData.getLockId();
          if (path.equals(LOCK_PATH)) {
            // clear all
            synchronized(classFaultMap) {
              classFaultMap.clear();
            }
          } else {
            InjectionInfo info = FaultHelper.convertAsInfo(path, lockService.getContents(path));

            if (info.fault == null) {
              LOG.debug("clear fault event [" + path + "]");
              internalClear(info.c, info.methodName);
            } else {
              LOG.debug("inject fault event [" + path + "], fault class :"
                  + info.fault.getClass().getName());
              internalInject(info.c, info.methodName, info.fault);
            }
          }
        } catch (IOException e) {
          LOG.warn("Fail to process event", e);
        }
      }
    };
  }

  private void internalClear(Class<?> c, String methodName) {
    synchronized (classFaultMap) {
      HashMap<String, Fault> faultMap = findFaultMap(c);

      if (faultMap != null) {
        faultMap.remove(methodName);
      }
      
      if (faultMap.isEmpty()) {
        classFaultMap.remove(c);
      }
      classFaultMap.notifyAll();
    }
  }

  private void internalInject(Class<?> c, String methodName, Fault fault) {
    synchronized (classFaultMap) {
      HashMap<String, Fault> faultMap = classFaultMap.get(c);
      if (faultMap == null) {
        faultMap = new HashMap<String, Fault>();
        classFaultMap.put(c, faultMap);
      }
      faultMap.put(methodName, fault);
      classFaultMap.notifyAll();
    }
  }

  private HashMap<String, Fault> findFaultMap(Class<?> c) {
    HashMap<String, Fault> ret = classFaultMap.get(c);

    if (ret == null) {
      Class<?>[] interfaces = c.getInterfaces();
      for (Class<?> ci : interfaces) {
        if ((ret = findFaultMap(ci)) != null) {
          return ret;
        }
      }

      return null;
    }

    return ret;
  }
  
  void waitUntil(Condition cond, long waitTime) throws IOException, InterruptedException {
    synchronized (classFaultMap) {
      long begin = System.currentTimeMillis();

      while (!cond.isSatisfied()) {
        classFaultMap.wait(waitTime);
        waitTime -= (System.currentTimeMillis() - begin);
        if (!cond.isSatisfied() && waitTime <= 0) {
          throw new IOException("fail to operation until timeout");
        }
      }
    }
  }
  
  static abstract class Condition {
    abstract boolean isSatisfied() throws IOException;
  }

  static class FaultHelper {
    static final String DELIMINATOR = "||";

    static class InjectionInfo {
      Class<?> c;
      String methodName;
      Fault fault;

      public String toString() {
        return "fault info, class : " + c.getName() + ", method : " + methodName + ", fault : "
            + ((fault == null) ? null : fault.getClass().getName());
      }
    }

    public static InjectionInfo convertAsInfo(String path, String contents) throws IOException {
      InjectionInfo info = new InjectionInfo();
      String[] pathElements = path.split("/");
      String className = null;

      try {
        String unConvertedClassName = pathElements[pathElements.length - 2];
        className = returnToClassName(unConvertedClassName);
        info.c = Class.forName(className);
      } catch (ClassNotFoundException e) {
        throw new IOException("class name is invalid [" + className + "]");
      }

      info.methodName = pathElements[pathElements.length - 1];

      if (contents.equals("")) {
        info.fault = null;
      } else {
        String faultName = null;
        try {
          faultName = returnToClassName(contents);
          info.fault = (Fault) Class.forName(faultName).newInstance();
        } catch (ClassNotFoundException e) {
          throw new IOException("class name is invalid [" + faultName + "]");
        } catch (Exception e) {
          throw new IOException("fail to instantiate fault", e);
        }
      }

      return info;
    }

    public static String convertAsContent(Class<?> c, String methodName, Fault fault) {
      return c.getName() + DELIMINATOR + methodName + DELIMINATOR + fault.getClass().getName();
    }

    public static String createLockPath(Class<?> c, String methodName) {
      return c.getName().replaceAll("\\.", "_") + "/" + methodName;
    }

    public static String returnToClassName(String convertedClassName) {
      return convertedClassName.replaceAll("_", "\\.");
    }
  }
}
